from pyspark.sql import Window
from pyspark.sql.session import SparkSession
from pyspark.sql.functions import *
spark= SparkSession.builder.appName('demo1_sparksession').enableHiveSupport().config("spark.jars", "jars/mysql-connector-java-8.0.29.jar").getOrCreate()
# /user/root/dm_db

staypoint_df = spark.read.format('csv').option('sep','\t').schema('mdn string,date string,county string,lon double,lat double,bsid string,grid_id string,biz_type string,event_type string,data_source string')\
    .load('/user/root/dm_db/staypoint')

usertag_df = spark.read.format('csv').option('sep',',').schema('mdn string,name string,gender string,age string,id_number string,number_attr string,trmnl_brand string,trmnl_price string,packg string,conpot string,resi_grid_id string,resi_county_id string')\
    .load('/user/root/dm_db/usertag')


# ##### 1.2 用户画像交叉分析（10分）
#
# - **分析维度**：
#   - 年龄分段(18-25,26-35,36-45,46+)
#   - 性别
#   - 消费潜力等级
# - **分析指标**：
#   - 各群体平均活动半径
#   - 跨区县活动比例
#   - 停留点时段分布


# ### 3. usertag 表
# | 字段名         | 数据类型 | 描述              |
# | -------------- | -------- | ----------------- |
# | mdn            | STRING   | 手机号大写MD5加密 |
# | name           | STRING   | 姓名              |
# | gender         | STRING   | 性别(1男2女)      |
# | age            | STRING   | 年龄              |
# | id_number      | STRING   | 证件号码          |
# | number_attr    | STRING   | 号码归属地        |
# | trmnl_brand    | STRING   | 终端品牌          |
# | trmnl_price    | STRING   | 终端价格          |
# | packg          | STRING   | 套餐              |
# | conpot         | STRING   | 消费潜力          |
# | resi_grid_id   | STRING   | 常住地网格        |
# | resi_county_id | STRING   | 常住地区县        |
#   - 停留点时段分布
staypoint_df = staypoint_df.join(usertag_df, 'mdn', 'inner') \
    .withColumn('age_range',
                when(col('age') <= 25, '18-25')
                .otherwise(when(col('age') <= 35, '26-35')
                           .otherwise(when(col('age') <= 45, '36-45')
                                      .otherwise('46+')))) \
    .withColumn('start_time', to_timestamp(split(col('date'), ',')[1], 'yyyyMMddHHmmss')) \
    .withColumn('end_time', to_timestamp(split(col('date'), ',')[0], 'yyyyMMddHHmmss')) \
    .withColumn('start_hour', hour(col('start_time'))) \
    .withColumn('end_hour', hour(col('end_time'))) \
    .withColumn('start-end-time',
                concat(hour(col('start_time')).cast('string'),
                       lit('-'),
                       hour(col('end_time')).cast('string')))
staypoint_df.show(truncate=False)


staypoint_df = staypoint_df.groupby('age_range', 'gender', 'conpot', 'start-end-time')\
    .agg(count(col('mdn')))

staypoint_df.show(truncate=False)

staypoint_df.write \
    .format("jdbc") \
    .option("url", "jdbc:mysql://master:3306") \
    .option("driver", "com.mysql.cj.jdbc.Driver") \
    .option("dbtable", "dianxim.demo_2") \
    .option("user", "root") \
    .option("password", "123456") \
    .mode("overwrite") \
    .save()

